package com.ruleengine.worker.runner.rule;

import static com.ruleengine.worker.exception.ExceptionMsg.UNSUPPORTED_RULE_TYPE;

import com.ruleengine.destination.DestinationManager;
import com.ruleengine.destination.IDestinationProvider;
import com.ruleengine.worker.exception.RuleRunnerException;
import com.ruleengine.worker.model.Rule;
import com.ruleengine.worker.runner.rule.compute.JsonFilterRuleRunner;
import com.ruleengine.worker.runner.rule.compute.JsonMapRuleRunner;
import com.ruleengine.worker.runner.rule.destination.PluginRuleRunner;
import com.ruleengine.worker.runner.rule.other.DispatchRuleRunner;
import com.ruleengine.worker.runner.rule.other.SimpleLogRuleRunner;

import cn.hutool.core.exceptions.UtilException;
import cn.hutool.core.util.ReflectUtil;
import io.vertx.core.Vertx;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lianghanwei on 2024/3/3
 */
@Slf4j
public class RuleRunnerFactory {

    private static final Vertx vertx = Vertx.vertx();

    public static RuleRunner newInstance(Rule rule, DestinationManager destinationManager) {

        switch (rule.getSpec().getType()) {
            case SIMPLE_LOG:
                return newInstance(SimpleLogRuleRunner.class, vertx, rule.getMetadata().getName(), rule);
            case DISPATCH:
                return newInstance(DispatchRuleRunner.class, vertx, rule.getMetadata().getName(), rule);
            case JSON_FILTER:
                return newInstance(JsonFilterRuleRunner.class, vertx, rule.getMetadata().getName(), rule);
            case JSON_MAP:
                return newInstance(JsonMapRuleRunner.class, vertx, rule.getMetadata().getName(), rule);
            case DEST_BIFROMQ:
            case DEST_KAFKA:
                return newInstance(PluginRuleRunner.class, vertx,
                        rule.getMetadata().getName(), rule, destinationManager);
            default:
                throw new RuleRunnerException(UNSUPPORTED_RULE_TYPE + ": ruleType=" + rule.getSpec().getType());
        }
    }

    private static RuleRunner newInstance(Class clazz, Object... args) {
        try {
            return (RuleRunner) ReflectUtil.newInstance(clazz, args);
        } catch (UtilException e) {
            log.error("Instance specified rule runner failed, className={}, args={}", clazz.getName(), args);
            throw new RuleRunnerException("Instance specified rule runner failed.", e);
        }
    }
}
